/*
 * main.c
 *
 *  Created on: 2011-7-14
 *      Author: rendong
 */
#include <stdio.h>
#include <pthread.h>

#include <netinet/in.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include<sys/epoll.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <getopt.h>
#include<errno.h>
#include <fcntl.h>

char * cfg_server_ip = "127.0.0.1";
int cfg_server_port = 3128;
int cfg_thread_num = 100;
int cfg_t = 0;
int msg_count = 0;

void write_addchannel(int fd);
void parse_arguments(int argc, char ** argv);
int connect_server(char *host, int port);
void *pthread_func (void *arg);

static void parse_message(int fd, short type, int len, char *message);
static void parse_auth_message(int fd, short type, int len, char *message);
static void parse_chat_message(int fd, short type, int len, char *message);
static void parse_online_message(int fd, short type, int len, char *message);

void write_auth_message(int fd);

void recv_message(int fd, char * buffer, int len);
ssize_t readn(int fd, void *ptr, size_t n);
void setnonblocking(int sock);


#define EPOLL_CLOSE_CODE(msg) do {\
ev.data.fd=sockfd;\
epoll_ctl(epfd,EPOLL_CTL_DEL,sockfd,&ev);\
close(sockfd);\
printf("client close fd:%d type:%s\n", sockfd, msg);\
events[i].data.fd=-1;\
}while(0);

#define WRITE_STR(pos, str, buffer) do{\
int str_len;\
str_len = strlen(str);\
memcpy(buffer+pos, &str_len, 4);\
pos += 4;\
memcpy(buffer + pos, str, str_len);\
pos += str_len;\
}while(0);

// 解析字符串
#define PARSE_STR(pmessage,pos,str_len, pstr) do{\
str_len = *(int *)(pmessage+(pos));\
pstr = malloc(str_len+1);\
memset(pstr, 0, str_len+1);\
memcpy(pstr, pmessage + (pos) + 4, str_len);\
pos += 4 + str_len;\
}while(0);

int cfg_buffer_size = 2048;

typedef struct _sgc_client_session_s
{
	int pos;
	int size;
	int remain;
	char * buffer;
}sgc_client_session_t;

char *buffers[65535] = {NULL};
int buffer_pos[65535] = {0};
int buffer_size[65535] = {0};
/**
 * main
 */
int main (int argc, char ** argv)
{
	parse_arguments(argc, argv);

	int epfd,i, sockfd,readyfds, nreadbytes;
	struct epoll_event ev, events[cfg_thread_num];

	epfd = epoll_create(cfg_thread_num);

	// 连接服务器
	for(i=0; i<cfg_thread_num; i++){
		sockfd = connect_server(cfg_server_ip, cfg_server_port);

		ev.data.fd = sockfd;
		ev.events=EPOLLIN|EPOLLET;

		epoll_ctl(epfd,EPOLL_CTL_ADD,sockfd,&ev);
		write_auth_message(sockfd);
	}
	pthread_t thread_id;
	pthread_create(&thread_id, NULL, pthread_func, NULL);

	// epoll 事件处理
	char * buffer = malloc(cfg_buffer_size);
	for(;;){
		readyfds=epoll_wait(epfd,events,cfg_thread_num,600);
		for(i=0; i<readyfds; i++){
			sockfd = events[i].data.fd;

			if(events[i].events & EPOLLIN)
			{
				memset(buffer, 0, cfg_buffer_size);
				while((nreadbytes = recv(sockfd, buffer, cfg_buffer_size, 0)) > 0)
				{
					//printf("read fd:%d len:%d\n", sockfd, nreadbytes);
					recv_message(sockfd, buffer, nreadbytes);
					memset(buffer, 0, cfg_buffer_size);
				}
				if (nreadbytes < 0){
					if (errno == ECONNRESET){
						EPOLL_CLOSE_CODE("connrset")
					}
				}else if (nreadbytes == 0){
					EPOLL_CLOSE_CODE("normal")
				}
			}else{
				EPOLL_CLOSE_CODE("exception")
			}
		}
	}

	return 0;
}

void *pthread_func (void *arg)
{
	int old_count;
	while(1)
	{
		if (old_count != msg_count){
			old_count = msg_count;
			printf("total msg_count:%d, thread:%d, per:%d\n", msg_count, cfg_thread_num, msg_count / cfg_thread_num);
		}
		sleep(1);
	}
	return NULL;
}

/**
 * 解析命令行参数
 */
void parse_arguments(int argc, char ** argv)
{
	char *l_opt_arg;
	struct option long_options[] = {
	{ "help",	0,	NULL,	'h'},
	{ "ip",		1,	NULL, 	'i'},
	{ "port",	1,	NULL,	'p'},
	{ "count",	1,	NULL,	'c'},
	{ "test",	1,	NULL,	't'},
	{  0,    0,    0,    0},
	};

	int c;
	while((c = getopt_long (argc, argv, "hi:p:c:t", long_options, NULL)) != -1)
	{
		switch (c)
		{
			case 'i':
				cfg_server_ip = optarg;
				break;
			case 'p':
				l_opt_arg = optarg;
				cfg_server_port = atoi(l_opt_arg);
				break;
			case 'c':
				l_opt_arg = optarg;
				cfg_thread_num = atoi(l_opt_arg);
				break;
			case 't':
				cfg_t = 1;
				break;
			case 'h':
			case '?':
				printf("Usage: %s [OPTION]...\n\
default ip=127.0.0.1 port=3128 count=100\n\n\
  -i, --ip=ip host        server ip host\n\
  -p, --port=socket port  server socket port num\n\
  -c, --count=thread num  client thread count\n\
  -t, --test              just connect\n\
  -h, --help              display this help and exit\n\n", argv[0]);
				exit(1);
				break;
		}
	}
	printf("start ip=%s port=%d count:%d\n", cfg_server_ip, cfg_server_port, cfg_thread_num);
}

/**
 * 连接服务器，返回socketfd
 */
int connect_server(char *host, int port)
{
	struct sockaddr_in client_addr;
	int client_socket_fd = -1;

	bzero(&client_addr,sizeof(client_addr));
	client_addr.sin_family = AF_INET;
	client_addr.sin_addr.s_addr = htons(INADDR_ANY); //INADDR_ANY表示自动获取本机地址
	client_addr.sin_port = htons(0); //0表示让系统自动分配一个空闲端口

	client_socket_fd = socket(AF_INET,SOCK_STREAM,0);

	if( client_socket_fd < 0){
		printf("Create Socket Failed! errno:%d err:%s\n", errno, strerror(errno));
		exit(1);
	}

	if( bind(client_socket_fd,(struct sockaddr*)&client_addr,sizeof(client_addr))){
		printf("Client Bind Port Failed!\n");
		exit(1);
	}

	struct sockaddr_in server_addr;
	bzero(&server_addr,sizeof(server_addr));
	server_addr.sin_family = AF_INET;
	if(inet_aton(host,&server_addr.sin_addr) == 0){
		printf("Server IP(%s) Address Error!\n", host);
		exit(1);
	}
	server_addr.sin_port = htons(port);
	socklen_t server_addr_length = sizeof(server_addr);

	if(connect(client_socket_fd,(struct sockaddr*)&server_addr, server_addr_length) < 0)
	{
		printf("Can Not Connect To %s:%d! err(%d)%s\n", host, port, errno, strerror(errno));
		exit(1);
	}

	setnonblocking(client_socket_fd);

	return client_socket_fd;
}

/**
 * 写入认证信息
 */
void write_auth_message(int fd)
{
	char * password = "password";
	char * userid = malloc(35);
	bzero(userid, 35);
	char * username = malloc(35);
	bzero(username, 35);

	sprintf(userid, "%s%d", "userid_", fd);
	sprintf(username, "%s%d", "username_", fd);

	printf("uid:%s username:%s\n", userid, username);
	int len,pos;
	short type;
	char * buffer;

	type = 100;
	len = 2 + 4 * 3 + strlen(userid)+strlen(username)+strlen(password);
	buffer = malloc(len+4);
	memcpy(buffer, &len, 4);
	memcpy(buffer+4, &type, 2);

	pos = 6;
	WRITE_STR(pos, userid, buffer)

	WRITE_STR(pos, username, buffer)

	WRITE_STR(pos, password, buffer)

	if (write(fd, buffer, len+4) < 0){
		printf("write error=%d %s\n", errno, strerror(errno));
	}
	free(buffer);
}


/**
 * 接收信息,分发信息
 */
void recv_message(int fd, char * buffer, int len)
{
	char * sbuffer = buffers[fd];
	if (sbuffer == NULL){
		sbuffer = malloc(len);

		buffer_pos[fd] = 0;
		buffer_size[fd] = len;
		buffers[fd] = sbuffer;

		memset(sbuffer, 0, len);
	}

	int remain = buffer_size[fd] - buffer_pos[fd];
	// 扩容
	if (remain < len){
		buffer_size[fd] = buffer_pos[fd] + len;
		char * tmp = malloc(buffer_size[fd]);
		memset(tmp, 0, buffer_size[fd]);
		memcpy(tmp, sbuffer, buffer_pos[fd]);
		free(buffers[fd]);
		buffers[fd] = tmp;
		sbuffer = tmp;
	}

	memcpy(sbuffer + buffer_pos[fd], buffer, len);
	buffer_pos[fd]+=len;

	int msglen, size;
	short type;
	char *message, *newbuffer;
	msglen = *(int *)sbuffer;// 消息长度包括类型长度+内容
	while(msglen > 0 && buffer_pos[fd] >= (msglen+4)){
		type = *(short *)(sbuffer + 4);

		message = malloc(msglen - 2);
		memset(message, 0, msglen -2);
		memcpy(message, sbuffer + 6, msglen - 2);

		size = buffer_pos[fd] - msglen-4;
		if (size < cfg_buffer_size)
			size = cfg_buffer_size;

		buffer_size[fd] = size;
		newbuffer = malloc(size);
		memset(newbuffer, 0, size);
		memcpy(newbuffer, sbuffer+msglen + 4, buffer_pos[fd] - msglen-4);

		free(sbuffer);
		buffers[fd] = newbuffer;
		sbuffer = newbuffer;
		buffer_pos[fd] = buffer_pos[fd] - (msglen+4);

		parse_message(fd, type, msglen -2, message);

		free(message);

		msglen = *(int*)sbuffer;
	}
	printf("fd=%d msglen:%d\n", fd, msglen);
}

// 解析消息，len为消息长度，不包括头长度
static void parse_message(int fd, short type, int len, char *message)
{
	msg_count++;
	switch(type){
	case 500:
		parse_auth_message(fd, type, len, message);
		break;
	case 502:
	case 503:
		parse_online_message(fd, type, len,message);
		break;
	case 506:
	case 507:
	case 508:
		parse_chat_message(fd, type, len, message);
		break;
	default:
		printf("parse_message, fd:%d type:%hd not found!\n", fd, type);
		break;
	}
}

static void parse_online_message(int fd, short type, int len, char *message)
{
	int uid_len;
	char * uid;
	int pos = 0;
	PARSE_STR(message, pos, uid_len, uid);

	if (type == 502){
		printf("fd=%d %s 上线\n", fd, uid);
	}else{
		printf("fd=%d %s 离线\n", fd, uid);
	}
}

static void parse_auth_message(int fd, short type, int len, char *message)
{
	int status = *(int *)message;
	printf("fd:%d, auth type:%d, status:%d\n", fd, type, status);

	write_addchannel(fd);
}

void write_addchannel(int fd)
{

}

static void parse_chat_message(int fd, short type, int len, char *message)
{
//len+type+srcUserID+srcUserName+name+message
//	长度+类型+发送用户编号+发送用户名+名称+接收消息内容
	int suid_len,sname_len,name_len,msg_len;
	char *suid,*sname,*name,*msg;
	int pos = 0;
	PARSE_STR(message,pos, suid_len, suid)
	PARSE_STR(message,pos, sname_len, sname)
	PARSE_STR(message,pos, name_len, name)
	PARSE_STR(message,pos, msg_len, msg)

	printf("fd=%d chat suid:%s sname:%s name:%s msg:%s\n", fd, suid, sname, name, msg);
	free(suid);
	free(sname);
	free(name);
	free(msg);
}


ssize_t             /* Read "n" bytes from a descriptor */
readn(int fd, void *ptr, size_t n)
{
    size_t       nleft;
    ssize_t      nread;

    nleft = n;
    while (nleft > 0) {
        if ((nread = read(fd, ptr, nleft)) < 0) {
            if (nleft == n)
                return(-1); /* error, return -1 */
            else
                break;      /* error, return amount read so far */
        } else if (nread == 0) {
            break;          /* EOF */
        }
        nleft -= nread;
        ptr = (char *)ptr + nread;
    }
    return(n - nleft);      /* return >= 0 */
}


/**
 * 设置非阻塞
 */
void setnonblocking(int sock)
{
    int opts;
    opts=fcntl(sock,F_GETFL);
    if(opts<0)
    {
    	printf("fcntl(sock,F_GETFL) error\n");
        exit(1);
    }

    opts = opts|O_NONBLOCK;

    if(fcntl(sock,F_SETFL,opts)<0)
    {
    	printf("fcntl(sock,SETFL,opts)");
        exit(1);
    }
}
